/*
* GPL
* Free soft,no warranted
* We are all in the mud, but some of us are looking at the stars.
* We are not born for ourselves.
*/
/**
* @file netconnection.hpp
* @Synopsis
* @author Chai JunJun
* @date 2009-07-10
*/

#ifndef QSERVER_SERVICE_CONNECTION_HPP__
#define QSERVER_SERVICE_CONNECTION_HPP__

#include <boost/assert.hpp>
#include <boost/any.hpp>
#include <boost/asio.hpp>
#include <boost/date_time/posix_time/posix_time_types.hpp>
#include <boost/bind.hpp>
#include <boost/noncopyable.hpp>
#include <boost/shared_ptr.hpp>
#include <boost/enable_shared_from_this.hpp>
#include <boost/function/function1.hpp>

#include "event.hpp"

namespace qserv 
{
    template<typename, typename, typename> class NetConnPool;
    template<typename, typename, typename> class NetService;

    /// Object for handle socket asynchronous operations.
    template<typename Work_Handler, typename Socket_Service = boost::asio::ip::tcp::socket>
    class NetConnection
        : public boost::enable_shared_from_this<NetConnection<Work_Handler, Socket_Service> >,
        private boost::noncopyable
    {
    public:
        using boost::enable_shared_from_this<NetConnection<Work_Handler, Socket_Service> >::shared_from_this;

        /// The type of the NetConnection.
        typedef NetConnection< Work_Handler, Socket_Service > netconnection_type;
        typedef boost::shared_ptr< netconnection_type > netconnection_ptr;

        /// The type of the work_handler.
        typedef Work_Handler work_handler_type;

        /// The type of the socket that will be used to provide asynchronous operations.
        typedef Socket_Service socket_type;

        typedef boost::function1< void, netconnection_ptr > createConnCommand;
        typedef boost::function1< void, netconnection_ptr > cancelConnCommand;

        typedef boost::shared_ptr<socket_type> socket_ptr;
        typedef boost::shared_ptr<work_handler_type> work_handler_ptr;
        typedef boost::shared_ptr<boost::asio::deadline_timer> timer_ptr;

        /// Construct a NetConnection object.
        explicit NetConnection(work_handler_type* work_handler,
            std::size_t timeout_seconds,
            std::size_t closed_wait_delay)
            : m_workHandlerPtr(work_handler),
            m_socketPtr(),
            m_timerPtr(),
            m_netIOServicePtr(0),
            m_workIOServicePtr(0),
            m_parentConn(0),
            m_childConn(0),
            m_timerCount(0),
            m_stoppedflag(true),
            m_closedflag(true),
            m_sendflag(false),
            m_timeout(timeout_seconds),
            m_closedWaitTime(boost::posix_time::seconds(closed_wait_delay)),
            m_restrictionTime(boost::posix_time::microsec_clock::universal_time())
        {
            BOOST_ASSERT(work_handler != 0);
            BOOST_ASSERT(closed_wait_delay != 0);
        }

        /// Destruct the service handler.
        ~NetConnection()
        {
            clearParentConnection();
            clearChildConnection();
        }

        /// Get the socket associated with the NetConnection.
        socket_type& socket()
        {
            BOOST_ASSERT(m_socketPtr.get() != 0);
            return *m_socketPtr;
        }


        /// Get the io_service object used to perform asynchronous operations.
        boost::asio::io_service& io_service()
        {
            BOOST_ASSERT(m_netIOServicePtr != 0);
            return *m_netIOServicePtr;
        }

        /// Get the io_service object used to dispatch synchronous works.
        boost::asio::io_service& work_service()
        {
            BOOST_ASSERT(m_workIOServicePtr != 0);
            return *m_workIOServicePtr;
        }

        /// Get the parent connection.
        boost::any* parentConnection()
        {
            return m_parentConn;
        }

        /// Get the child connection.
        boost::any* childConnection()
        {
            return m_childConn;
        }

        /// Close the connection with error_code e from any thread.
        void close(const boost::system::error_code& e)
        {
            // Dispatch to io_service thread.
            io_service().dispatch(boost::bind(&netconnection_type::close_i,
                shared_from_this(),
                e));
        }

        /// Close the handler with error_code 0 from any thread.
        void close()
        {
            close(boost::system::error_code(0, boost::system::get_system_category()));
        }

        /// Start an asynchronous operation from any thread to read any amount of data to buffers from the socket.
#if defined(__GNUC__)
        void async_read_some(boost::asio::mutable_buffers_1 buffers)
        {
            io_service().dispatch(boost::bind(&netconnection_type::async_read_some_1,
                shared_from_this(),
                buffers));
        }
#else  // other c++ compiler
        template<typename Buffers>
        void async_read_some(const Buffers& buffers)
        {
            io_service().dispatch(boost::bind(&netconnection_type::async_read_some_i<Buffers>,
                shared_from_this(),
                buffers));
        }
#endif

        void async_read(std::vector<unsigned char>& buffers)
        {
            async_read(boost::asio::buffer(buffers));
        }

        /// Start an asynchronous operation from any thread to read a certain amount of data to buffers from the socket.
#if defined(__GNUC__)
        void async_read(boost::asio::mutable_buffers_1 buffers)
        {
            io_service().dispatch(boost::bind(&netconnection_type::async_read_1,
                shared_from_this(),
                buffers));
        }
#else  // other c++ compiler
        template<typename Buffers>
        void async_read(const Buffers& buffers)
        {
            io_service().dispatch(boost::bind(&netconnection_type::async_read_i<Buffers>,
                shared_from_this(),
                buffers));
        }
#endif

        /// Start an asynchronous operation from any thread to write buffers to the socket.
#if defined(__GNUC__)
        void async_write(const boost::asio::const_buffers_1& buffers)
        {
            io_service().dispatch(boost::bind(&netconnection_type::async_write_1,
                shared_from_this(),
                buffers));
        }

        void async_write(const boost::asio::mutable_buffers_1& buffers)
        {
            io_service().dispatch(boost::bind(&netconnection_type::async_write_3,
                shared_from_this(),
                buffers));
        }

        void async_write(std::vector<boost::asio::const_buffer>& buffers)
        {
            io_service().dispatch(boost::bind(&netconnection_type::async_write_2,
                shared_from_this(),
                buffers));
        }

#else  // other c++ compiler
        template<typename Buffers>
        void async_write(const Buffers& buffers)
        {
            io_service().dispatch(boost::bind(&netconnection_type::async_write_i<Buffers>,
                shared_from_this(),
                buffers));
        }

#endif

        void async_write(boost::shared_ptr<event> event)
        {
            io_service().dispatch(boost::bind(&netconnection_type::async_write_i_2,
                shared_from_this(),
                event));
        }

        /// Dispatch event from the parent handler.
        void post_parent(boost::shared_ptr<event> event)
        {
            if (m_parentConn != 0)
            {
                work_service().post(boost::bind(&netconnection_type::do_parent,
                    shared_from_this(),
                    event));
            }
        }

        /// Dispatch event from the child handler.
        void post_child(boost::shared_ptr<event> event)
        {
            if (m_childConn != 0)
            {
                work_service().post(boost::bind(&netconnection_type::do_child,
                    shared_from_this(),
                    event));
            }
        }

    public:
        template<typename, typename, typename> friend class NetConnPool;
        template<typename, typename, typename> friend class NetService;
        template<typename, typename> friend class NetConnection;

        /// Set the parent handler.
        template<typename Parent_Handler>
        void parentConnection(Parent_Handler* handler)
        {
            clearParentConnection();

            if (handler != 0)
            {
                m_parentConn = new boost::any(handler);
            }
        }

        /// Set the child handler.
        template<typename Child_Handler>
        void childConnection(Child_Handler* handler)
        {
            clearChildConnection();

            if (handler != 0)
            {
                m_childConn = new boost::any(handler);
            }
        }

        /// Clear the parent handler.
        void clearParentConnection()
        {
            if (m_parentConn != 0)
            {
                delete m_parentConn;
                m_parentConn = 0;
            }
        }

        /// Clear the child handler.
        void clearChildConnection()
        {
            if (m_childConn != 0)
            {
                delete m_childConn;
                m_childConn = 0;
            }
        }

        void hookConnectionMade(createConnCommand cmd)
        {
            work_service().post(boost::bind(&netconnection_type::hookConnectionMade_i,
                shared_from_this(),
                cmd));
        }

        void hookConnectionCancel(cancelConnCommand cmd)
        {
            work_service().post(boost::bind(&netconnection_type::hookConnectionCancel_i,
                shared_from_this(),
                cmd));
        }

        void hookConnectionMade_i(createConnCommand cmd)
        {
            m_connCreateCommandQueue.push_back(cmd);
        }

        void hookConnectionCancel_i(cancelConnCommand cmd)
        {
            m_connCancelCommandQueue.push_back(cmd);
        }
        /// Check handler is in busy or not.
        bool is_busy()
        {
            return !m_closedflag || (boost::posix_time::microsec_clock::universal_time() < m_restrictionTime);
        }

        /// Bind a NetConnection with the given io_service and work_service.
        template<typename Work_Allocator>
        void bind(boost::asio::io_service& io_service,
            boost::asio::io_service& work_service,
            Work_Allocator* work_allocator)
        {
            BOOST_ASSERT(work_allocator != 0);
            BOOST_ASSERT(m_timerCount == 0);

            m_closedflag = false;
            m_stoppedflag = false;

            m_socketPtr.reset(work_allocator->make_socket(io_service));
            
            m_timerPtr.reset(new boost::asio::deadline_timer(io_service));
            //std::cout << m_timerPtr.use_count() << std::endl;
            m_netIOServicePtr = &m_socketPtr->get_io_service();
            m_workIOServicePtr = &work_service;

            // Clear work handler for new operations.
            m_workHandlerPtr->on_clear(*this);
        }

        /// Start an asynchronous connect.
        void connect(boost::asio::ip::tcp::endpoint& endpoint)
        {
            BOOST_ASSERT(m_socketPtr.get() != 0);
            BOOST_ASSERT(m_timerPtr.get() != 0);
            BOOST_ASSERT(m_workIOServicePtr != 0);

            // Set timer for connection timeout.
            // If reconnect, timer has been setted, don't set again.
            if (m_timerCount == 0)
                set_expiry(m_timeout);

            socket().lowest_layer().async_connect(endpoint,
                boost::bind(&netconnection_type::handle_connect,
                shared_from_this(),
                boost::asio::placeholders::error));
        }

        /// Start an asynchronous connect.
        void reconnect(boost::asio::ip::tcp::endpoint& endpoint)
        {
            BOOST_ASSERT(m_timerCount == 0);
            BOOST_ASSERT(m_socketPtr.get() != 0);
            m_closedflag = false;
            m_stoppedflag = false;

            m_timerPtr.reset(new boost::asio::deadline_timer(m_socketPtr->get_io_service()));

            // Clear work handler for new operations.
            m_workHandlerPtr->on_clear(*this);

            BOOST_ASSERT(m_timerPtr.get() != 0);
            BOOST_ASSERT(m_netIOServicePtr != 0);
            BOOST_ASSERT(m_workIOServicePtr != 0);

            // Set timer for connection timeout.
            // If reconnect, timer has been setted, don't set again.
            if (m_timerCount == 0)
                set_expiry(m_timeout);

            socket().lowest_layer().async_connect(endpoint,
                boost::bind(&netconnection_type::handle_connect,
                shared_from_this(),
                boost::asio::placeholders::error));
        }

        /// Start an asynchronous connect with the parent handler.
        template<typename Parent_Handler>
        void connect(boost::asio::ip::tcp::endpoint& endpoint,
            Parent_Handler* handler)
        {
            BOOST_ASSERT(handler != 0);

            BOOST_ASSERT(m_parentConn == 0);
            BOOST_ASSERT(m_childConn == 0);

            parentConnection(handler);
            handler->child_handler(this);

            connect(endpoint);
        }

        template<typename Parent_Handler>
        void reconnect(boost::asio::ip::tcp::endpoint& endpoint,
            Parent_Handler* handler)
        {
            BOOST_ASSERT(handler != 0);

            BOOST_ASSERT(m_parentConn == 0);
            BOOST_ASSERT(m_childConn == 0);

            parentConnection(handler);
            handler->child_handler(this);

            reconnect(endpoint);
        }
        /// Start the first operation.
        void start()
        {
            BOOST_ASSERT(m_socketPtr.get() != 0);
            BOOST_ASSERT(m_timerPtr.get() != 0);
            BOOST_ASSERT(m_workIOServicePtr != 0);

            // If start from connect, timer has been setted, don't set again.
            if (m_timerCount == 0)
                set_expiry(m_timeout);

            // Post to work_service for executing do_open.
            work_service().post(boost::bind(&netconnection_type::do_open,
                shared_from_this()));
        }

    private:
#if defined(__GNUC__)
        /// Start an asynchronous operation from io_service thread to read any amount of data to buffers from the socket.
        void async_read_some_1(const boost::asio::mutable_buffers_1& buffers)
        {
            async_read_some_i(buffers);
        }
#endif

        /// Start an asynchronous operation from io_service thread to read any amount of data to buffers from the socket.
        template<typename Buffers>
        void async_read_some_i(const Buffers& buffers)
        {
            // The handler has been stopped, do nothing.
            if (m_stoppedflag)
                return;
            socket().async_read_some(buffers,
                boost::bind(&netconnection_type::handle_read,
                shared_from_this(),
                boost::asio::placeholders::error,
                boost::asio::placeholders::bytes_transferred));
        }

#if defined(__GNUC__)
        /// Start an asynchronous operation from io_service thread to read a certain amount of data to buffers from the socket.
        void async_read_1(const boost::asio::mutable_buffers_1& buffers)
        {
            async_read_i(buffers);
        }
#endif

        /// Start an asynchronous operation from io_service thread to read a certain amount of data to buffers from the socket.
        template<typename Buffers>
        void async_read_i(const Buffers& buffers)
        {
            // The handler has been stopped, do nothing.
            if (m_stoppedflag)
                return;
            boost::asio::async_read(socket(),
                buffers,
                boost::bind(&netconnection_type::handle_read,
                shared_from_this(),
                boost::asio::placeholders::error,
                boost::asio::placeholders::bytes_transferred));
        }

#if defined(__GNUC__)
        /// Start an asynchronous operation from io_service thread to write buffers to the socket.
        void async_write_1(const boost::asio::const_buffers_1& buffers)
        {
            async_write_i(buffers);
        }

        void async_write_3(const boost::asio::mutable_buffers_1& buffers)
        {
            async_write_i(buffers);
        }

        void async_write_2(std::vector<boost::asio::const_buffer>& buffers)
        {
            async_write_i(buffers);
        }
#endif

        /// Start an asynchronous operation from io_service thread to write buffers to the socket.
        template<typename Buffers>
        void async_write_i(const Buffers& buffers)
        {
            // The handler has been stopped, do nothing.
            if (m_stoppedflag)
                return;
            boost::asio::async_write(socket(),
                buffers,
                boost::bind(&netconnection_type::handle_write,
                shared_from_this(),
                boost::asio::placeholders::error,
                boost::asio::placeholders::bytes_transferred));
        }

        void async_write_i_2(boost::shared_ptr<event> event)
        {
            // The handler has been stopped, do nothing.
            if (m_stoppedflag)
                return;
            m_recyclePool.push_back(event);
            if (!m_sendflag)
            {
                boost::asio::async_write(socket(),
                    boost::asio::buffer(event->getBuf(), event->getLen()),
                    boost::bind(&netconnection_type::handle_write_2,
                    shared_from_this(),
                    boost::asio::placeholders::error,
                    boost::asio::placeholders::bytes_transferred));
                m_sendflag = true;
            }
        }

        /// Set timer for connection.
        void set_expiry(std::size_t timeout_seconds)
        {
            if (timeout_seconds == 0)
                return;

            BOOST_ASSERT(m_timerPtr.get() != 0);

            ++m_timerCount;
            m_timerPtr->expires_from_now(boost::posix_time::seconds(timeout_seconds));
            m_timerPtr->async_wait(boost::bind(&netconnection_type::handle_timeout,
                shared_from_this(),
                boost::asio::placeholders::error));
        }

        /// Handle completion of a connect operation in io_service thread.
        void handle_connect(const boost::system::error_code& e)
        {
            // The handler has been stopped, do nothing.
            if (m_stoppedflag)
                return;

            if (!e)
            {
                start();
            }
            else
            {
                // Close with error_code e.
                close_i(e);
            }
        }

        /// Handle completion of a read operation in io_service thread.
        void handle_read(const boost::system::error_code& e,
            std::size_t bytes_transferred)
        {
            // The handler has been stopped, do nothing.
            if (m_stoppedflag)
                return;
            if (!e)
            {
                // Post to work_service for executing do_read.
                work_service().post(boost::bind(&netconnection_type::do_read,
                    shared_from_this(),
                    bytes_transferred));
            }
            else
            {
                // Close with error_code e.
                close_i(e);
            }
        }

        /// Handle completion of a write operation in io_service thread.
        void handle_write(const boost::system::error_code& e,
            std::size_t bytes_transferred)
        {
            // The handler has been stopped, do nothing.
            if (m_stoppedflag)
                return;

            if (!e)
            {
                // Post to work_service for executing do_write.
                work_service().post(boost::bind(&netconnection_type::do_write,
                    shared_from_this(),
                    bytes_transferred));
            }
            else
            {
                // Close with error_code e.
                close_i(e);
            }
        }

        /// Handle completion of a write operation in io_service thread.
        void handle_write_2(const boost::system::error_code& e,
            std::size_t bytes_transferred)
        {
            // The handler has been stopped, do nothing.
            if (m_stoppedflag)
                return;

            if (!e)
            {
                m_recyclePool.pop_front();
                if (m_recyclePool.empty())
                {
                    m_sendflag = false;
                }
                else
                {
                    boost::shared_ptr<event>& evt = m_recyclePool.front();
                    boost::asio::async_write(socket(),
                        boost::asio::buffer(evt->getBuf(), evt->getLen()),
                        boost::bind(&netconnection_type::handle_write_2,
                        shared_from_this(),
                        boost::asio::placeholders::error,
                        boost::asio::placeholders::bytes_transferred));
                }
                // Post to work_service for executing do_write.
                work_service().post(boost::bind(&netconnection_type::do_write,
                    shared_from_this(),
                    bytes_transferred));
            }
            else
            {
                // Close with error_code e.
                close_i(e);
            }
        }

        /// Handle timeout of whole operation in io_service thread.
        void handle_timeout(const boost::system::error_code& e)
        {
            --m_timerCount;

            // The connection has been stopped or timer has been cancelled, do nothing.
            if (m_stoppedflag || e == boost::asio::error::operation_aborted)
                return;

            if (!e)
            {
                //m_netIOServicePtr->stop();
                // Close with error_code boost::system::timed_out.
                close_i(boost::system::error_code(boost::asio::error::timed_out, boost::system::get_system_category()));
            }
            else
            {
                // Close with error_code e.
                //m_netIOServicePtr->stop();
                close_i(e);
            }
        }

        /// Close the handler in io_service thread.
        void close_i(const boost::system::error_code& e)
        {
            if (!m_stoppedflag)
            {
                m_stoppedflag = true;

                m_sendflag = false;
                m_recyclePool.clear();
                
                // Initiate graceful NetConnection closure.
                boost::system::error_code ignored_ec;
                socket().lowest_layer().shutdown(boost::asio::ip::tcp::socket::shutdown_both, ignored_ec);
                socket().lowest_layer().close();

                // Timer has not been expired, or expired but not dispatched,
                // cancel it even if expired.
                if (m_timerCount != 0)
                    m_timerPtr->cancel();

                // Post to work_service to executing do_close.
                work_service().post(boost::bind(&netconnection_type::do_close,
                    shared_from_this(),
                    e));
            }
        }

        /// Do on_open in work_service thread.
        void do_open()
        {
            // The handler has been stopped, do nothing.
            if (m_stoppedflag)
                return;

            createConnCommand cmd;
            BOOST_FOREACH(cmd, m_connCreateCommandQueue)
            {
                cmd(shared_from_this());
            }
            // Call on_open function of the work handler.
            m_workHandlerPtr->on_open(*this);
        }

        /// Do on_read in work_service thread.
        void do_read(std::size_t bytes_transferred)
        {
            // The handler has been stopped, do nothing.
            if (m_stoppedflag)
                return;
            // Call on_read function of the work handler.
            m_workHandlerPtr->on_read(*this, bytes_transferred);
        }

        /// Do on_write in work_service thread.
        void do_write(std::size_t bytes_transferred)
        {
            // The handler has been stopped, do nothing.
            if (m_stoppedflag)
                return;

            // Call on_write function of the work handler.
            m_workHandlerPtr->on_write(*this, bytes_transferred);
        }

        /// Do on_parent in work_service thread.
        void do_parent(boost::shared_ptr<event> event)
        {
            // The handler has been stopped, do nothing.
            if (m_stoppedflag)
                return;

            // Call on_parent function of the work handler.
            m_workHandlerPtr->on_parent(*this, event);
        }

        /// Do on_child in work_service thread.
        void do_child(boost::shared_ptr<event> event)
        {
            // The handler has been stopped, do nothing.
            if (m_stoppedflag)
                return;

            // Call on_child function of the work handler.
            m_workHandlerPtr->on_child(*this, event);
        }

        /// Do on_close and reset handler for next connaction in work_service thread.
        void do_close(const boost::system::error_code& e)
        {
            // Call on_close function of the work handler.
            m_workHandlerPtr->on_close(*this, e);

            // Clear parent handler and child handler.
            clearParentConnection();
            clearChildConnection();

            m_timerPtr.reset();

            // Set restriction time before reuse, wait uncompleted operations to be finished.
            m_restrictionTime = boost::posix_time::microsec_clock::universal_time() + m_closedWaitTime;

            m_closedflag = true;


            cancelConnCommand cmd;
            BOOST_FOREACH(cmd, m_connCancelCommandQueue)
            {
                cmd(shared_from_this());
            }
            m_connCreateCommandQueue.clear();
            m_connCancelCommandQueue.clear();
            // Leave socket to destroy delay for finishing uncompleted SSL operations.
            // Leave read_buffer_/write_buffer_/m_netIOServicePtr/m_workIOServicePtr for finishing uncompleted operations.
        }

    private:

        /// Work handler of the NetConnection.
        work_handler_ptr m_workHandlerPtr;

        /// Socket for the NetConnection.
        socket_ptr m_socketPtr;

        /// Timer for timeout operation.
        timer_ptr m_timerPtr;

        /// The io_service for for asynchronous operations.
        boost::asio::io_service* m_netIOServicePtr;

        /// The io_service for for executing synchronous works.
        boost::asio::io_service* m_workIOServicePtr;

        /// Parent handler of the handler.
        boost::any* m_parentConn;

        /// Child handler of the handler.
        boost::any* m_childConn;

        /// Count of waiting timer.
        std::size_t m_timerCount;

        // Flag to indicate that the handler has been stopped and can not do synchronous operations.
        bool m_stoppedflag;

        // Flag to indicate that the handler has been closed and can be reuse again after some seconds.
        bool m_closedflag;

        bool m_sendflag;
        /// The expiry seconds of connection.
        std::size_t m_timeout;

        /// For graceful close, if preallocated NetConnection number is not huge enough,
        /// should delay some seconds before any handler reuse again.
        boost::posix_time::time_duration m_closedWaitTime;

        /// The time of hander can use again.
        boost::posix_time::ptime m_restrictionTime;

        /// command invoke when connection start
        std::list< createConnCommand > m_connCreateCommandQueue;

        /// command invoke when connection close
        std::list< cancelConnCommand > m_connCancelCommandQueue;

        std::deque< boost::shared_ptr<event> > m_recyclePool;
    };

} // namespace qserv

#endif // QSERVER_SERVICE_CONNECTION_HPP__
